package com.shujia.opt

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo1ForeachPartition {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("partition")

    val sc = new SparkContext(conf)


    val studentRDD: RDD[String] = sc.textFile("data/stu")

    /**
      * 1、将创建链接放在Driver端，在Executor不能使用这个链接，网络链接不能在网络中传输
      * 2、将创建链接放在foreach里面，每一条数据都会创建一个链接，需要创建很多链接，效率低，而且会导致mysql链接不够
      *
      */

    /**
      * 将rdd的数据保存到mysql中
      *
      */

    /* studentRDD.foreach(line => {
       //写jdbc代码

       Class.forName("com.mysql.jdbc.Driver")
       //1、创建链接
       val con: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=utf-8", "root", "123456")

       //编写sql
       val stat: PreparedStatement = con.prepareStatement("insert into student(id,name,age,gender,clazz) values(?,?,?,?,?)")

       val split: Array[String] = line.split(",")

       //设置列值
       stat.setString(1, split(0))
       stat.setString(2, split(1))
       stat.setInt(3, split(2).toInt)
       stat.setString(4, split(3))
       stat.setString(5, split(4))


       //执行插入
       stat.execute()

     })*/

    /**
      * foreachPartition: 一次还进去一个分区的数据
      * 为每一个分区创建一个链接
      *
      *
      * 当我们将rdd的数据写入到外部数据库中的时候需要使用foreachPartition
      *
      */

    studentRDD.foreachPartition(iter => {

      println("创建链接")
      Class.forName("com.mysql.jdbc.Driver")
      //1、创建链接
      val con: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=utf-8", "root", "123456")

      iter.foreach(line => {
        //编写sql
        val stat: PreparedStatement = con.prepareStatement("insert into student(id,name,age,gender,clazz) values(?,?,?,?,?)")

        val split: Array[String] = line.split(",")

        //设置列值
        stat.setString(1, split(0))
        stat.setString(2, split(1))
        stat.setInt(3, split(2).toInt)
        stat.setString(4, split(3))
        stat.setString(5, split(4))


        //执行插入
        stat.execute()
      })

    })

  }

}
